.. _глава_84_start:

Параллелизм
============

.. index::
    parallel
    task
    asyncBuf
    map
    amap
    reduce

Большинство современных микропроцессоров состоит из более чем одного ядра, каждое из которых может
работать как отдельное вычислительное устройство. Они могут выполнять различные части
разных программ в одно и то же время. Возможности модуля std.parallelism
позволяют программам использовать преимущества всех ядер, чтобы
чтобы работать быстрее.
В этой главе рассматриваются следующие алгоритмы работы с диапазонами. Эти алгоритмы следует
использовать только в тех случаях, когда операции, которые должны выполняться параллельно, действительно
независимы друг от друга. Параллельно означает, что операции выполняются на
нескольких ядрах одновременно:
* parallel: Параллельный доступ к элементам диапазона.
* task: Создает задачи, которые выполняются параллельно.
* asyncBuf: параллельно итерирует элементы диапазона InputRange полу-жадно(??????)с
* map: Полупараллельно вызывает функции с элементами InputRange
* amap: Вызывает функции с элементами диапазона RandomAccessRange жадная функция (?????) параллельно.
* reduce: Выполняет вычисления над элементами RandomAccessRange параллельно
В программах, которые мы писали до сих пор, мы предполагали, что
выражения программы выполняются в определенном порядке, по крайней мере в общем случае построчно.
по строкам:

∙parallel: Accesses the elements of a range in parallel.
∙task: Creates tasks that are executed in parallel.
∙asyncBuf: Iterates the elements of an InputRange semi-eagerly in parallel. (????)
∙map: Calls functions with the elements of an InputRange semi-eagerly in
parallel.
∙amap: Calls functions with the elements of a RandomAccessRange fully-eagerly (????)
in parallel.
∙reduce: Makes calculations over the elements of a RandomAccessRange in

.. code-block:: d

    ++i;
    ++j;

В приведенном выше коде мы ожидаем, что значение i будет увеличено до того, как будет увеличено значение
j. Хотя семантически это правильно, в реальности так бывает редко.
реальности: микропроцессоры и компиляторы используют методы оптимизации, чтобы некоторые
переменные находятся в регистрах микропроцессора, которые независимы друг от
друг от друга. В этом случае микропроцессор будет выполнять операции, подобные
приращения выше, параллельно.
Хотя эти оптимизации эффективны, они не могут быть применены
автоматически к уровням выше, чем операции самого низкого уровня. Только
программист может определить, что определенные операции высокого уровня являются независимыми
и что они могут выполняться параллельно.
В цикле элементы диапазона обычно обрабатываются один за другим,
операции над каждым элементом следуют за операциями над предыдущими элементами:

.. code-block:: d

    auto students = [Student(1), Student(2), Student(3), Student(4)];
    foreach (student; students)
    {
        student.aSlowOperation();
    }


Обычно программа выполняется на одном из ядер
микропроцессора, которому операционная система поручила выполнение программы.
программы. Поскольку цикл foreach обычно оперирует элементами один за другим,
aSlowOperation() будет вызываться последовательно для каждого ученика. Однако во
многих случаях нет необходимости в том, чтобы операции предыдущих учеников были
перед началом операций с последующими учениками. Если бы операции
над объектами Student были действительно независимыми, было бы расточительно игнорировать
другие микропроцессорные ядра, которые потенциально могут простаивать в системе.
система.
Чтобы смоделировать длительные операции, в следующих примерах вызывается
Thread.sleep() из модуля core.thread. Thread.sleep() приостанавливает
операции на указанное время. По общему признанию, Thread.sleep - это
искусственный метод, используемый в следующих примерах, потому что он занимает время без
не загружая ни одно ядро. Несмотря на то, что это нереалистичный инструмент, он все же полезен в этой
Несмотря на нереалистичность, он все же полезен в этой главе для демонстрации возможностей параллелизма.

.. code-block:: d

    import std.stdio;
    import core.thread;

    struct Student
    {
        int number;
        void aSlowOperation()
        {
            writefln("The work on student %s has begun", number);
            // Wait for a while to simulate a long-lasting operation
            Thread.sleep(1.seconds);
            writefln("The work on student %s has ended", number);
        }
    }

    void main()
    {
        auto students =
            [Student(1), Student(2), Student(3), Student(4)];
        foreach (student; students)
        {
            student.aSlowOperation();
        }
    }

Время выполнения программы может быть измерено в терминале по времени:

.. code-block:: console

    $ time ./deneme
    The work on student 1 has begun
    The work on student 1 has ended
    The work on student 2 has begun
    The work on student 2 has ended
    The work on student 3 has begun
    The work on student 3 has ended
    The work on student 4 has begun
    The work on student 4 has ended
    real    0m4.005s ← 4 seconds total
    user    0m0.004s
    sys     0m0.000s

Поскольку итерация над учениками выполняется последовательно, а работа каждого
студента занимает 1 секунду, общее время выполнения составляет 4 секунды.
Однако если бы эти операции выполнялись в среде с 4 ядрами,
они могли бы выполняться одновременно, и общее время сократилось бы
примерно до 1 секунды.
Прежде чем посмотреть, как это делается, давайте сначала определим количество ядер, которые
доступных в системе, по параметру std.parallelism.totalCPUs:

.. index::
    std.parallelism
    taskPool

.. code-block::

    import std.stdio;
    import std.parallelism;

    void main()
    {
        writefln("There are %s cores on this system.", totalCPUs);
    }

Вывод программы в среде, в которой была написана эта глава
следующий:

.. code-block:: d

    There are 4 cores on this system.

taskPool.parallel()
--------------------

Эту функцию также можно вызвать просто как parallel().
parallel() осуществляет параллельный доступ к элементам диапазона. Эффективное применение -
в циклах foreach. Достаточно импортировать модуль std.parallelism и
заменить students на parallel(students) в приведенной выше программе достаточно
чтобы задействовать все ядра системы:

.. code-block:: d

    import std.parallelism;
    // ...
    foreach (student; parallel(students))
    {
    // ...
    }

Ранее в главе :ref:`foreach для структур и классов<глава_73_start>` мы видели, что
выражения, находящиеся в блоках foreach, передаются функциям-членам opApply() в качестве делегатов.
как делегаты. parallel() возвращает объект range, который знает, как
распределить выполнение делегата на отдельное ядро для каждого элемента.
В результате передача диапазона Student через parallel() заставляет приведенную выше программу
на системе с 4 ядрами завершается за 1 секунду:

.. code-block:: console

    $ time ./deneme
    The work on student 2 has begun
    The work on student 1 has begun
    The work on student 4 has begun
    The work on student 3 has begun
    The work on student 1 has ended
    The work on student 2 has ended
    The work on student 4 has ended
    The work on student 3 has ended

    real    0m1.005s ← now only 1 second
    user    0m0.004s
    sys     0m0.004s

Примечание: Время выполнения программы может отличаться на других системах, но оно
ожидается, что оно составит примерно "4 секунды, деленные на количество ядер".
Поток выполнения, проходящий через определенные части программы, называется потоком
или потоком. Программы могут состоять из нескольких потоков, которые
активно выполняются в одно и то же время. Операционная система запускает и выполняет каждый
ядра, а затем приостанавливает его для выполнения других потоков. Выполнение
каждого потока может включать множество циклов запуска и приостановки.
Все потоки всех программ, которые активны в данный момент времени, выполняются
выполняются на одних и тех же ядрах микропроцессора. Операционная система решает.
когда и при каких условиях запускать и приостанавливать каждый поток. Именно по этой
почему сообщения, выводимые aSlowOperation(), расположены в смешанном порядке
в приведенном выше выводе. Этот недетерминированный порядок выполнения потоков может
не иметь значения, если операции объектов Student действительно независимы друг от
друг от друга.
Программист обязан вызывать parallel() только тогда, когда
операции, применяемые к каждому элементу, независимы для каждой итерации. Например,
Например, если важно, чтобы сообщения появлялись в определенном порядке в
то вызов parallel() должен считаться ошибкой в приведенной выше программе.
Модель программирования, поддерживающая потоки, которые зависят от других потоков, называется
называется параллелизмом. Параллелизм - это тема следующей главы.
К моменту завершения параллельного цикла foreach все операции внутри цикла
были завершены для всех элементов. Программа может спокойно продолжить работу после
цикла foreach.

**Work unit size**

Второй параметр parallel() имеет перегруженное значение и игнорируется
в некоторых случаях:

.. code-block:: d

    /* ... */ = parallel(range, work_unit_size = 100);

При итерации по диапазонам RandomAccessRange:
Распределение потоков по ядрам имеет некоторую минимальную стоимость. Эта стоимость может
иногда может быть значительной, особенно если операции цикла завершаются за очень короткое время. В таких случаях может оказаться быстрее, если каждый поток выполнит более одной итерации цикла. Размер рабочей единицы
определяет количество элементов, которые должен выполнить каждый поток в каждой из
итераций:

.. code-block:: d

    foreach (student; parallel(students, 2))
    {
    // ...
    }

Значение по умолчанию для размера рабочей единицы равно 100 и подходит для большинства случаев.
* При итерации по диапазонам, не относящимся к RandomAccessRange:
parallel() не начинает параллельное выполнение, пока не будет выполнено количество элементов не-RandomAccessRange сначала последовательно. Из-за
относительно большого значения 100, параллельная() может создать неверное впечатление, что она неэффективна, когда ее пробуют на коротких
неслучайных диапазонах.
* При итерации по диапазонам результатов asyncBuf() или parallel map() (оба варианта
объясняются далее в этой главе):
Когда parallel() работает с результатами asyncBuf() или map(), он игнорирует
параметр размера рабочей единицы. Вместо этого parallel() повторно использует внутренний буфер
диапазона результатов.


.. index::
    task

Задача (Task)
-----------------

Операции, которые выполняются параллельно с другими операциями программы, называются
называются задачами. Задачи представлены типом std.parallelism.Task.
Фактически, parallel() создает новый объект Task для каждого рабочего потока и
запускает его автоматически. Затем parallel() ждет, пока все задачи не будут выполнены. перед окончательным выходом из цикла. parallel() очень удобна, поскольку она
конструирует, запускает и ожидает выполнения задач автоматически.
Когда задачи не соответствуют или не могут быть представлены элементами
диапазона, эти три шага могут быть выполнены программистом в явном виде. task()
конструирует, executeInNewThread() запускает, а yieldForce() ожидает объект задачи
объекта. Более подробно эти три функции описаны в комментариях к
следующей программе
В следующей программе функция anOperation() запускается дважды. Она
печатает первую букву id, чтобы указать, для какой задачи она работает.
*Примечание: Обычно символы, которые печатаются в потоки вывода типаestdout, не
не появляются на выходе сразу. Вместо этого они хранятся в буфере вывода до тех пор, пока не произойдет
пока не будет завершена строка вывода. Поскольку функция write не выводит символ новой строки, для того чтобы
чтобы наблюдать за параллельным выполнением следующей программы, вызывается stdout.flush()
чтобы отправить содержимое буфера в stdout еще до того, как будет достигнут конец строки.*

.. code-block:: d

    import std.stdio;
    import std.parallelism;
    import std.array;
    import core.thread;

    /* Печатает первую букву 'id' каждые полсекунды. Он
    * произвольно возвращает значение 1, чтобы имитировать функции, которые
    * производят вычисления. Этот результат будет использован позже в main. */
    int anOperation(string id, int duration)
    {
        writefln("%s will take %s seconds", id, duration);
        foreach (i; 0 .. (duration * 2))
        {
            Thread.sleep(500.msecs); /* half a second */
            write(id.front);
            stdout.flush();
        }
        return 1;
    }

    void main()
    {
        /* Создайте объект задачи, который будет выполнять
    * anOperation(). Параметры функции, которые
    * указанные здесь, передаются в функцию задачи как ее
    * параметры функции. */
        auto theTask = task!anOperation("theTask", 5);
        /* Запуск задачи */
        theTask.executeInNewThread();
        /* Пока 'theTask' продолжает выполняться, 'anOperation()'
    * вызывается снова, на этот раз непосредственно в main. */
        immutable result = anOperation("main's call", 3);
        /* На этом этапе мы уверены, что операция, которая была
    * была запущена непосредственно из main, была
    * завершена, потому что она была запущена обычным
    * вызовом функции, а не задачей. */
        /* С другой стороны, на данный момент неизвестно.
    * завершила ли 'theTask' свои операции
    * еще. yieldForce() ждет, пока задача завершит свои
    * операции; она возвращается только тогда, когда задача была
    * завершена. Ее возвращаемое значение - это возвращаемое значение
    * функции задачи, т. е. anOperation(). */
        immutable taskResult = theTask.yieldForce();
        writeln();
        writefln("All finished; the result is %s.",
            result + taskResult);
    }

Результат работы программы должен быть похож на следующий. Тот факт, что буквы m
и t выведены в смешанном порядке, указывает на то, что операции выполняются
параллельно:

.. code-block:: console

    main's call will take 3 seconds
    theTask will take 5 seconds
    mtmttmmttmmttttt
    All finished; the result is 2.

Функция задачи, приведенная выше, была указана в качестве шаблонного параметра task() как
task!anOperation. Хотя этот метод хорошо работает в большинстве случаев, как мы уже видели
как мы видели в главе :ref`Шаблоны<глава_64_start>`, каждая отдельная инстанция шаблона
шаблона является отдельным типом. Это различие может быть нежелательным в некоторых случаях.
ситуации, когда кажущиеся эквивалентными объекты задач на самом деле имеют разные
типы.
Например, хотя следующие две функции имеют одинаковую сигнатуру,
два инстанса Task, созданные в результате вызова шаблона функции task()
будут иметь разные типы. В результате они не могут быть
членами одного и того же массива:

.. code-block:: d

    import std.parallelism;
    double foo(int i)
    {
        return i * 1.5;
    }

    double bar(int i)
    {
        return i * 2.5;
    }

    void main()
    {
        auto tasks = [task!foo(1),
            task!bar(2)]; // ← compilation ERROR
    }

.. code-block:: console

    Error: incompatible types for ((task(1)) : (task(2))):
    'Task!(foo, int)*' and 'Task!(bar, int)*'

Другой перегруженный вариант функции task() принимает функцию в качестве первого параметра функции
вместо этого:

.. code-block:: d

    void someFunction(int value)
    {
    // ...
    }
    auto theTask = task(&someFunction, 42)


Поскольку этот метод не предполагает различных инстанций шаблона Task, он
позволяет помещать такие объекты в один массив:

.. code-block:: d

    import std.parallelism;

    double foo(int i)
    {
        return i * 1.5;
    }

    double bar(int i)
    {
        return i * import std.parallelism;

        double foo(int i)
        {
            return i * 1.5;
        }

        double bar(int i)
        {
            return i * 2.5;
        }

        void main()
        {
            auto tasks = [task(&foo, 1),
                task(&bar, 2)]; // ← compiles
        }

        2.5;
    }

    void main()
    {
        auto tasks = [task(&foo, 1),
            task(&bar, 2)]; // ← compiles
    }

Лямбда-функция или объект типа, определяющего член opCall, также могут быть использованы
использоваться в качестве функции задачи. В следующем примере запускается задача, выполняющая функцию
лямбда

.. code-block:: d

    auto theTask = task((int value) {
        /* ... */
    }, 42);


**Исключения**

Поскольку задачи выполняются в отдельных потоках, исключения, которые они вызывают, не могут
не могут быть пойманы потоком, который их запустил. По этой причине исключения, которые
автоматически перехватываются самими задачами, чтобы быть перехваченными позже.
при вызове функций-членов Task, таких как yieldForce(). Это позволяет
что позволяет главному потоку перехватывать исключения, вызванные задачей.

.. code-block:: d

    import std.stdio;
    import std.parallelism;
    import core.thread;

    void mayThrow()
    {
        writeln("mayThrow() is started");
        Thread.sleep(1.seconds);
        writeln("mayThrow() is throwing an exception");
        throw new Exception("Error message");
    }

    void main()
    {
        auto theTask = task!mayThrow();
        theTask.executeInNewThread();
        writeln("main is continuing");
        Thread.sleep(3.seconds);
        writeln("main is waiting for the task");
        theTask.yieldForce();
    }

Вывод программы показывает, что не пойманное исключение, которое было
брошенное задачей, не завершает всю программу сразу (оно
завершает только задачу):


.. code-block:: console

    main is continuing
    mayThrow() is started
    mayThrow() is throwing an exception ← thrown
    main is waiting for the task
    object.Exception@deneme.d(10): Error message ← terminate

yieldForce() можно вызвать в блоке try-catch, чтобы перехватить исключения, которые
которые выбрасывает задача. Обратите внимание, что это отличается от однопоточных задач: В однопоточных
однопоточных программах, как в примерах, которые мы писали до этой главы,
try-catch оборачивает код, который может выбросить исключение. В параллельных программах он оборачивает
yieldForce():

.. code-block:: d

    try
    {
        theTask.yieldForce();
    }
    catch (Exception exc)
    {
        writefln("Detected an error in the task: '%s'", exc.msg);
    }

На этот раз исключение перехватывается главным потоком, и не приводит к завершению
программы:

.. code-block:: console

    main is continuing
    mayThrow() is started
    mayThrow() is throwing an exception ← thrown
    main is waiting for the task
    Detected an error in the task: 'Error message' ← caught

**Member functions of Task**

* done: Возвращает статус задачи : завершена или нет. Повторно выбрасывет исключение если задача завершилась с выбросом исключения

.. code-block:: d

    if (theTask.done)
    {
        writeln("Yes, the task has been completed");
    }
    else
    {
        Parallelism 611 writeln("No, the task is still going on");
    }

* executeInNewThread(int priority): Запускает задачу в новом потоке с
  указанным приоритетом. (Приоритет - это понятие операционной системы, определяющее
  приоритеты выполнения потоков.)
  Есть три функции для ожидания завершения задачи:
* yieldForce(): Запускает задачу, если она еще не была запущена; если она уже
  завершена, возвращает возвращаемое значение; если задача еще выполняется, ожидает ее
  завершения, не заставляя микропроцессор быть занятым; если было выброшено исключение
  было выброшено, повторно выбрасывает это исключение.
* SpinForce(): Работает аналогично yieldForce(), за исключением того, что заставляет
  микропроцессор занят во время ожидания, чтобы поймать завершение как можно раньше
  как можно раньше.
* workForce(): Работает аналогично yieldForce(), за исключением того, что запускает новую
  задачу в текущем потоке в ожидании завершения задачи.
  В большинстве случаев yieldForce() является наиболее подходящей функцией для вызова при ожидании
  Приостанавливает поток, вызвавший yieldForce(), до тех пор, пока задача не будет выполнена.
  пока задача не будет завершена. Хотя функция spinForce() заставляет микропроцессор быть занятым во время
  в ожидании, она подходит, когда ожидается, что задача будет выполнена в очень короткое время.
  workForce() может быть вызвана, когда запуск других задач предпочтительнее, чем
  приостановки текущего потока.
  О других функциях-членах Task читайте в онлайн-документации Phobos.Task.

taskPool.asyncBuf()
-------------------
Аналогично parallel(), asyncBuf() параллельно итерирует диапазоны InputRange. Она
сохраняет элементы в буфере по мере их получения диапазоном и выдает элементы
элементы из этого буфера своему пользователю.
Чтобы избежать превращения потенциально полностью ленивого диапазона ввода в полностью энергичный (???? fully-eagerly)
диапазон, он выполняет итерацию элементов волнами. Как только он подготовит определенное количество
элементов параллельно, он ждет, пока эти элементы не будут потреблены функцией popFront()
прежде чем выдать элементы следующей волны.
asyncBuf() принимает диапазон и необязательный размер буфера, который определяет, сколько
сколько элементов будет доступно во время каждой волны:

.. code-block:: d

    auto elements = taskPool.asyncBuf(range, buffer_size);

Чтобы увидеть эффект от использования asyncBuf(), давайте воспользуемся диапазоном, которому требуется полсекунды для
итерации и полсекунды на обработку каждого элемента. Этот диапазон просто производит
целые числа до указанного предела:

.. code-block:: d

    import core.thread;

    struct Range
    {
        int limit;
        int i;
        bool empty() const
        {
            return i >= limit;
        }

        int front() const
        {
            return i;
        }

        void popFront()
        {
            writefln("Producing the element after %s", i);
            Thread.sleep(500.msecs);
            ++i;
        }
    }

    void main()
    {
        auto range = Range(10);
        foreach (element; range)
        {
            writefln("Using element %s", element);
            Thread.sleep(500.msecs);
        }
    }

Элементы производятся и используются лениво. Поскольку на каждый элемент уходит одна секунда
элемент, весь диапазон обрабатывается в этой программе за десять секунд:

.. code-block:: console

    $ time ./deneme
    Using element 0
    Producing the element after 0
    Using element 1
    Producing the element after 1
    Using element 2
    ...
    Producing the element after 8
    Using element 9
    Producing the element after 9
    real    0m10.007s ← 10 seconds total
    user    0m0.004s
    sys     0m0.000s

В соответствии с этим выводом элементы производятся и используются последовательно.
С другой стороны, может не быть необходимости ждать, пока предыдущие элементы будут обработаны
перед началом производства последующих элементов. Программа
займет меньше времени, если другие элементы могут быть произведены, пока передний элемент находится в процессе
пока передний элемент используется:

.. code-block:: d

    import std.parallelism;
    //...
    foreach (element; taskPool.asyncBuf(range, 2))
    {
    }

В приведенном выше вызове asyncBuf() готовит два элемента в своем буфере. Элементы
производятся параллельно, пока они используются:

.. code-block:: console

    $ time ./deneme
    Producing the element after 0
    Producing the element after 1
    Using element 0
    Producing the element after 2
    Using element 1
    Producing the element after 3
    Using element 2
    Producing the element after 4
    Using element 3
    Producing the element after 5
    Using element 4
    Producing the element after 6
    Producing the element after 7
    Using element 5
    Using element 6
    Producing the element after 8
    Producing the element after 9
    Using element 7
    Using element 8
    Using element 9
    real    0m6.007s ← now 6 seconds
    user    0m0.000s
    sys     0m0.004s

По умолчанию размер буфера равен 100. Размер буфера, обеспечивающий наилучшую
производительности, будет отличаться в разных ситуациях.
asyncBuf() можно использовать и вне циклов foreach. Например.
следующий код использует возвращаемое значение asyncBuf() в качестве InputRange, который
который работает в полусвободном режиме (?????semi-eagerly):

.. code-block:: d

    auto range = Range(10);
    auto asyncRange = taskPool.asyncBuf(range, 2);
    writeln(asyncRange.front);

taskPool.map()
--------------

Прежде чем объяснять map() из модуля std.algorithm, необходимо объяснить
taskPool.map(). std.algorithm.map - это алгоритм, часто встречающийся во многих
функциональных языках. Он вызывает функцию с элементами диапазона один за другим
и возвращает диапазон, состоящий из результатов вызова этой функции с каждым
элементом. Это ленивый алгоритм: Он вызывает функцию по мере необходимости. (Существует также
std.algorithm.each, который предназначен для генерации побочных эффектов для каждого элемента, в отличие от
в отличие от получения результата из него.)
Тот факт, что std.algorithm.map работает лениво, является очень сильным во многих
программах. Однако если функция все равно должна вызываться с каждым элементом
и операции над каждым элементом независимы друг от друга, ленивость
может оказаться неоправданно медленнее, чем параллельное выполнение. taskPool.map() и
taskPool.amap() из модуля std.parallelism используют преимущества
нескольких ядер и во многих случаях работают быстрее.
Давайте сравним эти три алгоритма на примере Student. Предположим.
что у Student есть функция-член, которая возвращает среднюю оценку
студента. Чтобы продемонстрировать, насколько параллельные алгоритмы быстрее, давайте снова замедлим работу этой
функцию с помощью Thread.sleep().
std.algorithm.map принимает функцию в качестве параметра шаблона, а
range в качестве параметра функции. Она возвращает диапазон, состоящий из результатов
применения этой функции к элементам диапазона:

.. code-block:: d

    auto result_range = map!func(range);

Функция может быть задана синтаксисом => как лямбда-выражение, как мы видели в предыдущих главах.
Следующая программа использует map() для вызова функции-члена averageGrade() для каждого элемента:

.. code-block:: d

    import std.stdio;
    import std.algorithm;
    import core.thread;

    struct Student
    {
        int number;
        int[] grades;
        double averageGrade()
        {
            writefln("Started working on student %s",
                number);
            Thread.sleep(1.seconds);
            const average = grades.sum / grades.length;
            writefln("Finished working on student %s", number);
            return average;
        }
    }

    void main()
    {
        Student[] students;
        foreach (i; 0 .. 10)
        {
            /* Two grades for each student */
            students ~= Student(i, [80 + i, 90 + i]);
        }
        auto results = map!(a => a.averageGrade)(students);
        foreach (result; results)
        {
            writeln(result);
        }
    }

Вывод программы демонстрирует, что map() работает лениво;
averageGrade() вызывается для каждого результата по мере итерации цикла foreach:

.. code-block:: console

    $ time ./deneme
    Started working on student 0
    Finished working on student 0
    85 ← calculated as foreach iterates
    Started working on student 1
    Finished working on student 1
    86
    ...
    Started working on student 9
    Finished working on student 9
    94
    real    0m10.006s ← 10 seconds total
    user    0m0.000s
    sys     0m0.004s

Если бы std.algorithm.map был жадным (eager) алгоритмом, то сообщения о начале и завершении операций
и завершении операций выводились бы в самом верху.
TaskPool.map() из модуля std.parallelism работает по сути так же, как и std.algorithm.map.
так же, как и std.algorithm.map. Единственное отличие заключается в том, что он выполняет функции
и сохраняет результаты в буфере, из которого они могут быть получены по мере необходимости.
Размер этого буфера определяется вторым параметром. Например.
следующий код подготовит результаты вызовов функций для трех
элементов за один раз:

.. code-block:: d

    import std.parallelism;

    // ...
    double averageGrade(Student student)
    {
        return student.averageGrade;
    }
    // ...
    auto results = taskPool.map!averageGrade(students, 3);


*Примечание:Вышеуказанная функция free-standingaverageGrade() необходима из-за ограничения
которое связано с использованием локальных делегатов с шаблонами функций-членов, такими как
TaskPool.map. Без этой отдельно стоящей функции произошла бы ошибка компиляции
функции:*

.. code-block:: d

    auto results = taskPool.map!(a => a.averageGrade)(students, 3); // ← compilation ERROR

На этот раз операции выполняются этапами по три элемента:


.. code-block:: console

    $ time ./deneme
    Started working on student 1 ← in parallel
    Started working on student 2 ← but in unpredictable order
    Started working on student 0
    Finished working on student 1
    Finished working on student 2
    Finished working on student 0
    85
    86
    87
    Started working on student 4
    Started working on student 5
    Started working on student 3
    Finished working on student 4
    Finished working on student 3
    Finished working on student 5
    88
    89
    90
    Started working on student 7
    Started working on student 8
    Started working on student 6
    Finished working on student 7
    Finished working on student 6
    Finished working on student 8
    91
    92
    93
    Started working on student 9
    Finished working on student 9
    94
    real    0m4.007s ← 4 seconds total
    user    0m0.000s
    sys     0m0.004s

Второй параметр map() имеет то же значение, что и asyncBuf(): Он
определяет размер буфера, в котором map() хранит результаты. Третий
является размером рабочей единицы(???? unit size), как и в parallel(); разница в том, что по умолчанию его значение
который по умолчанию равен size_t.max:

.. code-block:: d

    /* ... */ = taskPool.map!func(range,
    buffer_size = 100
    work_unit_size = size_t.max);

Поскольку эта функция жадная (eager), все результаты будут готовы к моменту возвращения функции amap().

taskPool.amap()
---------------

Параллельная amap() работает так же, как и параллельная map(), с двумя отличиями:
* Это полностью eager. (жадная ?????)
* Работает с диапазонами RandomAccessRange.
auto results = taskPool.amap!averageGrade(students);
Поскольку эта функция работает в режиме eager, все результаты будут готовы к моменту возврата amap():

.. code-block:: console

    $ time ./deneme
    Started working on student 1 ← all are executed up front
    Started working on student 0
    Started working on student 2
    Started working on student 3
    Finished working on student 1
    Started working on student 4
    Finished working on student 2
    Finished working on student 3
    Started working on student 6
    Finished working on student 0
    Parallelism
    616
    Started working on student 7
    Started working on student 5
    Finished working on student 4
    Started working on student 8
    Finished working on student 6
    Started working on student 9
    Finished working on student 7
    Finished working on student 5
    Finished working on student 8
    Finished working on student 9
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    real    0m3.005s ← 3 seconds total
    user    0m0.000s
    sys     0m0.004s


amap() работает быстрее, чем map(), за счет использования массива, который достаточно велик.
для хранения всех результатов. Для увеличения скорости он потребляет больше памяти.
Необязательный второй параметр amap() также является размером рабочей единицы (work unit????):

.. code-block:: d

    auto results = taskPool.amap!averageGrade(students, 2);

Результаты также можно хранить в RandomAccessRange, который передается в amap()
в качестве третьего параметра:

.. code-block:: d

    double[] results;
    results.length = students.length;
    taskPool.amap!averageGrade(students, 2, results);

taskPool.reduce()
------------------

Как и в случае с map(), сначала стоит объяснить reduce() из модуля std.algorithm
Во-первых.
reduce() - это эквивалент std.algorithm.fold, который мы уже рассматривали
ранее в главе :ref:`Диапазоны<глава_81_start>`. Основное различие между этими двумя функциями заключается в том.
что параметры их функций меняются местами. (По этой причине я рекомендую
предпочесть fold() для непараллельного кода, поскольку она может использовать преимущества :ref:`UFCS<глава_61_start>`) в цепочечных выражениях диапазона).
reduce() - еще один высокоуровневый алгоритм, часто встречающийся во многих функциональных
языках. Как и map(), он принимает одну или несколько функций в качестве параметров шаблона.
В качестве параметров функции принимается значение, которое будет использоваться в качестве начального значения результата
и диапазон. reduce() вызывает функции с текущим значением
результата и каждого элемента диапазона. Если начальное значение не указано, вместо него используется первый
вместо него используется первый элемент диапазона.
Если предположить, что в ее реализации определена переменная с именем result, то
Работа функции reduce() может быть описана следующими шагами:
#. Присваивает начальное значение result
#. Выполняет выражение result = func(result, element) для каждого
элемента
#. Возвращает конечное значение result
Например, сумма квадратов элементов массива может быть вычислена
как в следующей программе:


.. code-block:: d

    import std.stdio;
    import std.algorithm;

    void main()
    {
        writeln(reduce!((a, b) => a + b * b)(0, [5, 10]));
    }

Когда функция задается синтаксисом =>, как в приведенной выше программе, первый
параметр (здесь a) представляет собой текущее значение результата (инициализированное параметром
параметром 0 выше), а второй параметр (здесь b) представляет текущий
элемент.
Программа выводит сумму 25 и 100, квадраты 5 и 10:

Вывод:

.. code-block:: console

    125

Как видно из его поведения, reduce() использует цикл в своей реализации.
Поскольку этот цикл обычно выполняется на одном ядре, он может быть неоправданно
медленным, если вызовы функций для каждого элемента независимы друг от друга.
В таких случаях можно использовать taskPool.reduce() из модуля std.parallelism.
для использования преимуществ всех ядер.
Чтобы увидеть пример этого, давайте используем reduce() с функцией, которая искусственно замедлена:

.. code-block:: d

    import std.stdio;
    import std.algorithm;
    import core.thread;

    int aCalculation(int result, int element)
    {
        writefln("started  - element: %s, result: %s",
            element, result);
        Thread.sleep(1.seconds);
        result += element;
        writefln("finished - element: %s, result: %s",
            element, result);
        return result;
    }

    void main()
    {
        writeln("Result: ", reduce!aCalculation(0, [1, 2, 3, 4]));
    }

reduce() использует элементы  последовательно, чтобы получить конечное значение результата:

.. code-block:: console

    $ time ./deneme
    started  - element: 1, result: 0
    finished - element: 1, result: 1
    started  - element: 2, result: 1
    finished - element: 2, result: 3
    started  - element: 3, result: 3
    finished - element: 3, result: 6
    started  - element: 4, result: 6
    finished - element: 4, result: 10
    Result: 10
    real    0m4.003s ← 4 seconds total
    user    0m0.000s
    sys     0m0.000s

Как и в примерах parallel() и map(), необходимо импортировать модуль std.parallelism
и вызова taskPool.reduce() достаточно, чтобы использовать преимущества всех
ядер:

.. code-block:: d

    import std.parallelism;
    // ...
    writeln("Result: ", taskPool.reduce!aCalculation(0, [1, 2, 3, 4]))

Однако в работе taskPool.reduce() есть важные отличия.
Как и другие параллельные алгоритмы, taskPool.reduce() выполняет функции
параллельно, используя элементы в разных задачах. Каждая задача работает с элементами
которые ей назначены, и вычисляет результат, соответствующий элементам
этой задачи. Поскольку reduce() вызывается только с одним начальным значением, каждая задача должна
использовать это же начальное значение для инициализации своего результата (параметр 0 выше).
Конечные значения результатов, которые производит каждая задача, сами используются в
вычислении того же результата в последний раз. Эти окончательные вычисления выполняются
последовательно, а не параллельно. По этой причине taskPool.reduce() может выполняться
медленнее в коротких примерах, как в этой главе, что видно из следующего
выводе.
Тот факт, что одно и то же начальное значение используется для всех задач, фактически
используется в вычислениях несколько раз, taskPool.reduce() может вычислить
результат, отличный от того, что вычислит std.algorithm.reduce(). По этой
поэтому начальное значение должно быть идентичным значению для выполняемого вычисления.
например, 0 в данном примере, которое не оказывает никакого влияния на сложение.
Кроме того, поскольку результаты используются одними и теми же функциями в последний раз в
последовательных вычислениях, типы параметров, которые принимают функции, должны быть
быть совместимы с типами значений, которые эти функции возвращают.
taskPool.reduce() следует использовать только при соблюдении следующих условий

.. code-block:: d

    import std.parallelism;
    // ...
    writeln("Result: ", taskPool.reduce!aCalculation(0, [1, 2, 3, 4]))

Вывод программы показывает, что сначала вычисления выполняются параллельно
параллельно, а затем их результаты вычисляются последовательно. Вычисления, которые
выполняются последовательно, выделены цветом:

.. code-block:: console

    $ time ./deneme
    started  - element: 3, result: 0 ← first, the tasks in parallel
    started  - element: 2, result: 0
    started  - element: 1, result: 0
    started  - element: 4, result: 0
    finished - element: 3, result: 3
    finished - element: 1, result: 1
    started  - element: 1, result: 0 ← then, their results sequentially
    finished - element: 4, result: 4
    finished - element: 2, result: 2
    finished - element: 1, result: 1
    started  - element: 2, result: 1
    finished - element: 2, result: 3
    started  - element: 3, result: 3
    finished - element: 3, result: 6
    started  - element: 4, result: 6
    finished - element: 4, result: 10
    Result: 10
    real    0m5.006s ← parallel reduce is slower in this example
    user    0m0.004s
    sys     0m0.000s

Параллельное reduce() быстрее во многих других вычислениях, например при вычислении
математической константы pi(π) в квадратурах.

Несколько функций и результаты кортежей
----------------------------------------------

std.algorithm.map(), taskPool.map(), taskPool.amap() и
taskPool.reduce() могут принимать более одной функции, в этом случае
результаты возвращаются в виде кортежа. Мы уже встречались с типом Tuple в главе :ref:`Кортежи<глава_77_start>`. 
Результаты отдельных функций соответствуют
элементам кортежа в том порядке, в котором указаны функции. Например,
результат первой функции - это первый член кортежа.
Следующая программа демонстрирует работу нескольких функций с
std.algorithm.map. Обратите внимание, что возвращаемые типы функций не обязательно должны быть одинаковыми
как это показано в функциях quarterOf() и tenTimes() ниже. В этом случае
типы членов кортежей также будут разными:

.. code-block:: d

    import std.stdio;
    import std.algorithm;
    import std.conv;

    double quarterOf(double value)
    {
        return value / 4;
    }

    string tenTimes(double value)
    {
        return to!string(value * 10);
    }

    void main()
    {
        auto values = [10, 42, 100];
        auto results = map!(quarterOf, tenTimes)(values);
        writefln(" Quarters  Ten Times");
        foreach (quarterResult, tenTimesResult; results)
        {
            writefln("%8.2f%8s", quarterResult, tenTimesResult);
        }
    }


Вывод :

.. code-block:: console

    Quarters Ten Times
    2.50 100
    10.50 420
    25.00 1000

В случае taskPool.reduce() начальные значения результатов должны быть
в виде кортежа:

.. code-block:: d

    taskPool.reduce!(foo, bar)(tuple(0, 1), [1, 2, 3, 4]);


TaskPool
---------

За кулисами все параллельные алгоритмы из модуля std.parallelism
используют объекты задач, которые являются элементами контейнера TaskPool. Обычно все
алгоритмы используют один и тот же объект-контейнер с именем taskPool.
TaskPool содержит соответствующее количество задач в зависимости от
окружения, в котором работает программа. По этой причине обычно нет необходимости
необходимости создавать какой-либо другой объект TaskPool. Тем не менее, явные объекты TaskPool могут
могут быть созданы и использованы по мере необходимости.
Конструктор TaskPool принимает количество потоков для использования во время
параллельных операций, которые впоследствии запускаются через него. По умолчанию
количество потоков на один меньше, чем количество ядер в системе. Все
все возможности, которые мы рассмотрели в этой главе, могут быть применены к отдельному объекту TaskPool
объекту TaskPool.
В следующем примере вызывается функция parallel() для локального объекта TaskPool:


.. code-block:: d

    import std.stdio;
    import std.parallelism;

    void main()
    {
        auto workers = new TaskPool(2);
        foreach (i; workers.parallel([1, 2, 3, 4]))
        {
            writefln("Working on %s", i);
        }
        workers.finish();
    }

TaskPool.finish() указывает объекту прекратить обработку, когда все его текущие
задачи завершены.

Резюме
------

* Ошибкой является параллельное выполнение операций, если эти операции не являются
  не являются независимыми друг от друга.
* parallel() осуществляет параллельный доступ к элементам диапазона.
* Задачи можно явно создавать, запускать и ожидать с помощью task(),
  executeInNewThread() и yieldForce() соответственно.
* Исключения, которые выходят из задач, могут быть пойманы позже большинством функций
  функций параллелизма, таких как yieldForce().
* asyncBuf() параллельно итерирует элементы InputRange с нетерпением.
* map() вызывает функции с элементами InputRange полуэкспериментально
  параллельно.
* amap() вызывает функции с элементами RandomAccessRange полностью-
  параллельно.
* reduce() выполняет вычисления над элементами RandomAccessRange параллельно
  параллельно.
* map(), amap() и reduce() могут принимать несколько функций и возвращать
  результаты в виде кортежей.
* При необходимости можно использовать объекты TaskPool, отличные от taskPool.
